package com.hhf.rrd.udf.two;

import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.functions.ScalarFunction;
import org.apache.flink.types.Row;

import java.math.BigDecimal;

/**
 * // todo
 *
 * @author huanghaifeng15
 * @date 2022/2/15 20:06
 **/
public class UserScalarFunction extends ScalarFunction {


    /**
     * 自定义数据类型作为输出参数
     */
    @DataTypeHint("ROW<age INT, name STRING, totalBalance DECIMAL(10, 2)>")
    public Row eval(long i) {
        User u = new User();
        if (i > 0 && i <= 5) {
            u.age = (int) i;
            u.name = "name1";
            u.totalBalance = new BigDecimal("1.1");
        } else {
            u.age = (int) i;
            u.name = "name2";
            u.totalBalance = new BigDecimal("2.2");
        }

        return Row.of(u.age,u.name,u.totalBalance);
    }

    /**
     * 自定义数据类型作为输入参数
     */
    public String eval(@DataTypeHint("ROW<age INT, name STRING, totalBalance DECIMAL(10, 2)>") Row i) {
        assert i.getField(0) != null;
        assert i.getField(0) != null;
        int age = Integer.parseInt(i.getField(0).toString());
        if (age > 0 && age <= 5) {
            User u = new User();
            u.age = 1;
            u.name = "name1";
            u.totalBalance = new BigDecimal("1.1");
            return u.name;
        } else {
            User u = new User();
            u.age = 2;
            u.name = "name2";
            u.totalBalance = new BigDecimal("2.2");
            return u.name;
        }
    }
}
